package com.techblor.redis.library.config;

import com.techblor.redis.library.annotation.RedisSubscribe;
import com.techblor.redis.library.message.listener.RedisMessageListenerInfo;
import com.techblor.redis.library.message.proxy.RedisMessageListenerInvocationHandler;
import com.techblor.redis.library.message.handler.RedisMessageErrorHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.util.ReflectionUtils;

import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

@Slf4j
@Configuration
public class RedisMessageListenerConfig implements BeanPostProcessor {

    private RedisMessageListenerContainer container = new RedisMessageListenerContainer();

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        scanRedisSubscribe(bean);
        return bean;
    }

    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
        container.setConnectionFactory(connectionFactory);
        container.setErrorHandler(new RedisMessageErrorHandler());
        return container;
    }


    /**
     * 扫描redis订阅注解并开始监听频道
     *
     * @param bean
     */
    private void scanRedisSubscribe(Object bean) {
        List<RedisMessageListenerInfo> listeners = getRedisMessageListenerInfo(bean);
        addRedisMessageListener(listeners);
    }

    /**
     * 获取监听器
     *
     * @param bean
     * @return
     */
    public List<RedisMessageListenerInfo> getRedisMessageListenerInfo(Object bean) {
        Method[] methods = ReflectionUtils.getAllDeclaredMethods(bean.getClass());
        return Arrays.stream(methods).filter(method -> method.isAnnotationPresent(RedisSubscribe.class))
                .map(method -> {
                    RedisSubscribe redisSubscribe = method.getAnnotation(RedisSubscribe.class);
                    return RedisMessageListenerInfo.builder()
                            .object(bean)
                            .topics(Arrays.stream(redisSubscribe.value()).map(ChannelTopic::new).collect(Collectors.toList()))
                            .method(method)
                            .build();
                }).collect(Collectors.toList());
    }


    public void addRedisMessageListener(List<RedisMessageListenerInfo> listeners) {
        for (RedisMessageListenerInfo info : listeners) {
            try {
                log.info("开始监听频道: {}", info.getTopics());
                MessageListener messageListener = (MessageListener) Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[]{MessageListener.class},
                        new RedisMessageListenerInvocationHandler(info));
                container.addMessageListener(messageListener, info.getTopics());
            } catch (Exception e) {
                log.error(e.getMessage());
            }
        }
    }

}
